package cn.cosmosx.base.lock;

import cn.cosmosx.base.exception.BaseException;
import cn.cosmosx.base.exception.PlatformException;
import cn.cosmosx.base.service.RedisService;
import cn.cosmosx.base.spel.ExpressionEvaluator;
import cn.cosmosx.base.util.DateUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.concurrent.BasicThreadFactory;
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.annotation.Around;
import org.aspectj.lang.annotation.Aspect;
import org.aspectj.lang.reflect.MethodSignature;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.expression.AnnotatedElementKey;
import org.springframework.expression.EvaluationContext;
import org.springframework.stereotype.Component;

import java.util.Iterator;
import java.util.UUID;
import java.util.concurrent.*;

/**
 * 分布式锁处理器
 */
@Slf4j
@Aspect
@Component
public class LockHandlerAspect implements InitializingBean {
    @Autowired
    private RedisService redisService;
    ExpressionEvaluator<String> evaluator = new ExpressionEvaluator<>();
    /**
     * 扫描的任务队列
     */
    private final static ConcurrentLinkedQueue<LockDefinitionHolder> HOLDERS = new ConcurrentLinkedQueue<>();
    /**
     * 线程池，维护keyAliveTime
     */
    private static final ScheduledExecutorService SCHEDULER = Executors.newSingleThreadScheduledExecutor(new BasicThreadFactory.Builder().namingPattern("lock-schedule-pool").daemon(true).build());

    @Around("@annotation(lock)")
    public Object around(ProceedingJoinPoint joinPoint, Lock lock) {
        String businessKey = getBusinessKey(joinPoint, lock);
        String uniqueValue = UUID.randomUUID().toString();
        // 加锁
        try {
            if (!redisService.setIfAbsent(businessKey, uniqueValue)) {
                log.error("当前业务:{}正在流程中，不能重复提交", businessKey);
                throw new PlatformException("当前业务正在执行中，请勿重复操作[0]");
            }
            redisService.expire(businessKey, lock.lockTime());
            Thread currentThread = Thread.currentThread();
            // 将本次 Task 信息加入「延时」队列中
            HOLDERS.add(new LockDefinitionHolder(businessKey, lock.lockTime(), System.currentTimeMillis(), currentThread, lock.tryCount()));
            // 执行业务操作
            Object result = joinPoint.proceed();
            // 线程被中断，抛出异常，中断此次请求
            if (currentThread.isInterrupted()) {
                log.error("当前业务:{}异常中止", businessKey);
                throw new PlatformException("网络繁忙，请稍后再试[1]");
            }
            return result;
        } catch (Throwable e) {
            log.error("has some error, please check again", e);
            if (e instanceof BaseException) {
                throw new PlatformException(((BaseException) e).getResultCode(), e.getMessage());
            }
            throw new PlatformException("网络繁忙，请稍后再试[2]");
        } finally {
            log.info("release the lock, businessKey is [" + businessKey + "]");
            // 请求结束后，强制删掉 key，释放锁
            redisService.delete(businessKey);
        }
    }

    /**
     * 获取BusinessKey
     *
     * @param jp
     * @param lock
     * @return
     */
    private String getBusinessKey(ProceedingJoinPoint jp, Lock lock) {
        EvaluationContext context = evaluator.createEvaluationContext(jp);
        AnnotatedElementKey elementKey = evaluator.createAnnotatedElementKey(((MethodSignature) jp.getSignature()).getMethod(), jp.getTarget().getClass());
        String value = evaluator.getValue(lock.value(), elementKey, context, String.class);
        return String.format("%s::%s", lock.prefix(), value);
    }

    @Override
    public void afterPropertiesSet() {
        init();
    }

    /**
     * 启动化定时器
     */
    void init() {
        // 两秒执行一次「续时」操作
        SCHEDULER.scheduleAtFixedRate(() -> {
            // 这里记得加 try-catch，否者报错后定时任务将不会再执行
            try {
                Iterator<LockDefinitionHolder> iterator = HOLDERS.iterator();
                while (iterator.hasNext()) {
                    LockDefinitionHolder holder = iterator.next();
                    log.info("当前线程:{}-业务key:{}-加锁时长(s):{}-上次更新时间:{}-总尝试次数:{}-当前尝试次数{}-更新时间周期(ms):{}",
                            holder.getCurrentTread(), holder.getBusinessKey(), holder.getLockTime(),
                            DateUtil.format(holder.getLastModifyTime()),
                            holder.getTryCount(), holder.getCurrentCount(), holder.getModifyPeriod());
                    // 判断 key 是否还有效，无效的话进行移除
                    if (redisService.get(holder.getBusinessKey()) == null) {
                        iterator.remove();
                        continue;
                    }
                    // 超时重试次数，超过时给线程设定中断
                    if (holder.getCurrentCount() > holder.getTryCount()) {
                        holder.getCurrentTread().interrupt();
                        iterator.remove();
                        continue;
                    }
                    // 判断是否进入最后三分之一时间
                    long curTime = System.currentTimeMillis();
                    boolean shouldExtend = (holder.getLastModifyTime() + holder.getModifyPeriod()) <= curTime;
                    if (shouldExtend) {
                        holder.setLastModifyTime(curTime);
                        redisService.expire(holder.getBusinessKey(), holder.getLockTime());
                        log.info("businessKey : [" + holder.getBusinessKey() + "], try count : " + holder.getCurrentCount());
                        holder.setCurrentCount(holder.getCurrentCount() + 1);
                    }
                }
            } catch (Exception e) {
                log.error("lock pools execute error", e);
            }
        }, 0, 2, TimeUnit.SECONDS);
    }
}
